package com.mjf.gmall.realtime.dim

import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.JSON
import com.mjf.gmall.realtime.bean.dim.UserInfo
import com.mjf.gmall.realtime.util.{MyKafkaUtil, OffsetManager}
import org.apache.hadoop.conf.Configuration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.phoenix.spark._

/**
 * 实时维护用户信息维度表
 *
 * 技术栈：
 *  zookeeper/kafka/redis/hadoop/HBase/phoenix
 *
 * phoenix建表语句：
 *  create table gmall0105_user_info(id varchar primary key, user_level varchar, birthday varchar, gender varchar, age_group varchar, gender_name varchar) SALT_BUCKETS = 3;
 */
object UserInfoApp {
  def main(args: Array[String]): Unit = {
    // 读取Kafka中用户的topic
    // 加载流
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("user_info_app")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))

    val topic = "ods_user_info"
    val groupId = "user_info_group"

    // 从redis读取offset
    val kafkaOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topic, groupId)

    // 从Kafka消费数据
    var recordInputStream: InputDStream[ConsumerRecord[String, String]] = null
    if (kafkaOffsetMap == null && kafkaOffsetMap.size >0) {
      // 如果redis中已经有了offset信息
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, kafkaOffsetMap, groupId)
    } else {
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)
    }

    //获得偏移结束点
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]  // 写在算子外面为启动时执行
    val inputGetOffsetDstream: DStream[ConsumerRecord[String, String]] = recordInputStream.transform {
      rdd =>
        // 获得本批次的偏移量的结束位置，用于更新redis中的偏移量
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  // 写在算子里面是周期性执行（driver端）
        rdd
    }


    // 写入到HBase中
    inputGetOffsetDstream.foreachRDD{
      rdd =>
        val userInfoRDD: RDD[UserInfo] = rdd.map {
          record =>
            val jsonString: String = record.value()
            val userInfo: UserInfo = JSON.parseObject(jsonString, classOf[UserInfo])
            val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
            val birthday: Date = sdf.parse(userInfo.birthday)
            val curTs: Long = System.currentTimeMillis()
            val betweenTs: Long = curTs - birthday.getTime
            val age: Long = betweenTs / 1000L / 60L / 60L / 24L / 365L

            if(age < 20L) {
              userInfo.age_group = "20岁及以下"
            } else if(age > 30L) {
              userInfo.age_group = "30岁以上"
            } else {
              userInfo.age_group = "21岁到30岁"
            }

            if(userInfo.gender == "M") {
              userInfo.gender_name = "男"
            } else {
              userInfo.gender_name = "女"
            }
            userInfo
        }

        userInfoRDD.saveToPhoenix(
          "GMALL0105_USER_INFO",
          Seq("ID", "USER_LEVEL", "BIRTHDAY", "GENDER", "AGE_GROUP", "GENDER_NAME"),
          new Configuration,
          Some("hadoop102,hadoop103,hadoop104:2181")
        )

        // 数据处理结束，偏移量提交到redis
        OffsetManager.saveOffset(topic, groupId, offsetRanges)
    }

    ssc.start()
    ssc.awaitTermination()
  }
}
